package com.dongzili.binlog.common.config.redis;

import com.dongzili.binlog.common.binlogsynchronization.BinLogItem;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import javax.annotation.Resource;
import java.time.Duration;

@Slf4j
@Configuration
@AutoConfigureAfter(RedisConfig.class)
public class StreamMessageListenerConfig {

    public final static String streamKey = "binlog-stream";
    public final static String group = "binlog-group";
    public final static String consumer = "binlog-consumer";


    @Resource
    private RedisTemplate<String, BinLogItem> redisTemplate;



    @Bean
    public StreamMessageListenerContainer<String, ObjectRecord<String, BinLogItem>> streamMessageListenerContainer(
            RedisConnectionFactory connectionFactory, RedisStreamConsumer messageConsumer) {

        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, BinLogItem>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .pollTimeout(Duration.ofMillis(100))
                        .targetType(BinLogItem.class)
                        .build();

        StreamMessageListenerContainer<String, ObjectRecord<String, BinLogItem>> listenerContainer =
                StreamMessageListenerContainer.create(connectionFactory, options);

        try {
            // 创建消费者组
            messageConsumer.redisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), group);
        } catch (Exception e) {
            // 消费者组可能已经存在
            log.error("redis stream group已存在 {} {}", streamKey, group);
        }

        listenerContainer.receive(
                Consumer.from(group, consumer),
                StreamOffset.create(streamKey, ReadOffset.lastConsumed()),
                messageConsumer);

        listenerContainer.start();

        return listenerContainer;
    }


//
//    @Bean
//    public StreamMessageListenerContainer<String, ObjectRecord<String, BinLogItem>> streamMessageListenerContainer(
//            RedisConnectionFactory connectionFactory, RedisStreamConsumer messageConsumer) {
//
//        try {
//            // 创建消费者组
//            messageConsumer.getRedisTemplate().opsForStream().createGroup(streamKey, ReadOffset.from("0-0"), groupName);
//        } catch (Exception e) {
//            // 消费者组可能已经存在
//        }
//        try{
//            redisTemplate.opsForStream().createGroup(binLogConstants.getServerKey(), StreamMessageListenerConfig.group);
//        }catch (Exception e){
//            log.error("redis stream group已存在 {} {}", binLogConstants.getServerKey(), StreamMessageListenerConfig.group);
//        }
//
//        // 用于配置消息监听容器的选项。在这个方法中，通过设置不同的选项，如轮询超时时间和消息的目标类型，可以对消息监听容器进行个性化的配置。
//        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, BinLogItem>> options =
//                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
//                        .builder()
//                        // 设置了轮询超时的时间为100毫秒。这意味着当没有新的消息时，容器将每隔100毫秒进行一次轮询。
//                        .pollTimeout(Duration.ofMillis(100))
//                        // 指定了消息的目标类型为 String。这意味着容器会将接收到的消息转换为 String 类型，以便在后续的处理中使用。
//                        .targetType(BinLogItem.class)
//                        .build();
//
//        // 创建一个可用于监听Redis流的消息监听容器。
//        StreamMessageListenerContainer<String, ObjectRecord<String, BinLogItem>> listenerContainer =
//                StreamMessageListenerContainer.create(connectionFactory, options);
//
//        // 方法配置了容器来接收来自特定消费者组和消费者名称的消息。它还指定了要读取消息的起始偏移量，以确定从哪里开始读取消息。
//        listenerContainer.receive(
//                Consumer.from(StreamMessageListenerConfig.group, consumer),
//                StreamOffset.create(binLogConstants.getServerKey(), ReadOffset.lastConsumed()),
//                messageConsumer);
//
//        // 方法启动了消息监听容器，使其开始监听消息。一旦容器被启动，它将开始接收并处理来自Redis流的消息。
//        listenerContainer.start();
//
//        return listenerContainer;
//    }

}
